package peer

import (
	"../util"
	"bytes"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
)

//定义节点结构
type Peer struct {
	peerID   int64            //当前节点的id
	f        int64            //节点个数
	N        int64            //节点总数
	k        int64            //设置的稳定检查点更新区间
	reqNo    int64            //当前处理的分配的请求序号
	peerList map[int64]string //string->url
	view     int64            //当前视图
	//preSave   map[uint]*premessage	//
	reqStore    map[string]*RequestMsgs //string对应批量请求的摘要，Requests对应批请求,这里指的是接受到的客户端请求消息不是节点之间三阶段的请求
	msgStore    map[msgIndex]*msgSave   //通过v和n定位一条消息，其他节点发送的消息自己接受到后需要持久化，这里存储的是其他节点发送的三阶段请求消息，只有在消息完成prePrepare阶段才会被存储到这个map
	lastExecNum int                     //节点完成的最后一个请求序列号
	//prePreSet map					    //已经完成的预准备消息集合
	//preSet							//已经完成的准备消息集合
	MsgEntrance chan interface{} //收到消息
	MsgDelivery chan interface{} //向其他节点广播消息
	//reqStore    map[string]req		//存储接受到的批量请求  string为该请求的hash,req为请求
	h int64 //低水线
	H int64 //高水线
}

//RequestMsgs是收到的交易请求的结合
type RequestMsgs struct {
	reqs []*RequestMsg
}

//根据v和n定位一条prepare消息
type msgIndex struct {
	v int64
	n int64
}

//根据摘要和n定位于一条prepare消息
type prepqreIndex struct {
	d string
	n uint
}

//存储接受到的每条消息
type msgSave struct {
	digest      string         //消息摘要
	prePrepare  *PrePrepareMsg //接受到主节点的消息存储
	sentPrepare bool           //是否已经发送Prepare消息
	prepare     []*PrepareMsg  //每一个prepare消息中可能存储不同类型的消息，因此验证收到不同的消息的个数的时候会通过v,n
	commit      []*CommitMsg   //保存从其他节点收到的commit消息
	sendCommit  bool           //是否已经发出commit消息
}

type Server struct {
	url  string
	peer *Peer
}

//判断自己是不是主节点
func (p *Peer) primaryID(view int64) (primaryID int64) {
	primaryID = view % (p.N)
	return
}

//三阶段需要多少个消息才能达成共识
func (p *Peer) intersectionQuorum() int64 {
	return (p.N + p.f + 2) / 2
}

func (p *Peer) judgeByNum(num int64) bool {
	return num >= p.intersectionQuorum()
}

//判断是不是达到接受消息数量， 比如3f+1个节点需要2f+1

//根据消息摘要，视图，序列号判断一条消息有没有达成完成prepared状态
func (p *Peer) prePrepared(d string, v int64, n int64) bool {
	//判断该消息是否存在消息请求存储中,reqStore是只要有请求过来就会存储。
	//if _, ok := p.reqStore[d]; ok {
	//	return false
	//}
	//如果在存储三阶段中收到的消息map中能根据v和n找到该记录，表明该记录已经完成prePrepare阶段
	if msg, ok := p.msgStore[msgIndex{v, n}]; ok && msg != nil {
		pre := msg.prePrepare
		fmt.Println(pre.ViewID,pre.SequenceID,pre.Digest)
		fmt.Println(v,n,d)
		if pre.ViewID == v && pre.SequenceID == n && pre.Digest == d {

			fmt.Println("确认完成preprepare阶段")
			return true
		}else {
			fmt.Println("hahhahahahhha")
		}
	}
	return false
}

//判断消息是否完成了prepared阶段
func (p *Peer) prepared(d string, v int64, n int64) bool {
	fmt.Println("测试主节点1")
	//消息还没有完成prePrepare阶段
	if !p.prePrepared(d, v, n) {
		return false
	}
	//存在已经完成prepared集合中，视图相同，摘要相同，那么一定完成了prepared阶段
	fmt.Println("测试主节点2")
	//如果在存储中没有存储该消息，那么
	msg := p.msgStore[msgIndex{v, n}]
	if msg == nil {
		return false
	}

	//统计从其他节点收到多少条消息\
	var receiveMsgNum int64 = 0
	for _, prepareMsg := range msg.prepare {
		if prepareMsg.ViewID == v && prepareMsg.SequenceID == n && prepareMsg.BatchDigest == d {
			receiveMsgNum++
		}
	}
	fmt.Println(receiveMsgNum)
	//这里加1,是因为收不到主节点的prepare消息
	return p.judgeByNum(receiveMsgNum+1)

}

func (p *Peer) commited(d string, v int64, n int64) bool {
	//消息还没有完成prepared阶段
	if !p.prepared(d, v, n) {
		return false
	}

	//统计其他节点收到多少条消息\
	msg := p.msgStore[msgIndex{v, n}]
	if msg == nil {
		return false
	}

	var receiveMsgNum int64 = 0
	for _, commitMsg := range msg.commit {
		if commitMsg.ViewID == v && commitMsg.SequenceID == n {
			receiveMsgNum++
		}
	}

	return p.judgeByNum(receiveMsgNum)
}

//判断是否达到发送commit消息条件
func (p *Peer) maybeSendCommit(d string, v int64, n int64) {
	//从存储三阶段消息的存储中找到消息
	fmt.Println("maycommit")
	msg := p.getMsgSave(v, n)
	//如果该消息已经完成perpared阶段,该消息还没有发送commit类型消息
	if p.prepared(d, v, n) && !msg.sendCommit {
		//创建commit消息
		commitMsg := &CommitMsg{
			ViewID:     v,
			SequenceID: n,
			Digest:     d,
			ReplicaId:  p.peerID,
		}
		msg.sendCommit = true
		//自己也要处理commit消息

		fmt.Println("从节点广播commit消息了")
		//为了能接收到新的请求，索引这里暂时假设消息已经被处理，那么需要序列号++
		p.reqNo++
		fmt.Println("下一个处理请求的序列号是",p.reqNo)

		//第一次广播消息之前需要将消息添加到自己的日志记录中
		p.getCommitMsg(commitMsg)
		//广播commit消息

		p.MsgDelivery <- commitMsg
	}
	return

}

//获取本节点已经收到的三阶段中的消息存储
func (p *Peer) getMsgSave(v, n int64) (msg *msgSave) {
	msgIndex := msgIndex{v, n}
	msg, ok := p.msgStore[msgIndex]
	if ok {
		return
	}
	msg = &msgSave{}
	p.msgStore[msgIndex] = msg
	return
}

//广播请求给其他节点
func (p *Peer) Broadcast() {
	for {
		msg := <-p.MsgDelivery
		switch msg.(type) {
		case *RequestMsg:
			for peerID, url := range p.peerList {
				//防止发消息给自己
				if peerID == p.peerID {
					continue
				}

				jsonMsg, err := json.Marshal(msg)

				if err != nil {
					log.Panic(err)
				}
				send(url+"/req", jsonMsg)
			}

		case *PrePrepareMsg:
			for peerID, url := range p.peerList {
				//防止发消息给自己
				if peerID == p.peerID {
					continue
				}

				jsonMsg, err := json.Marshal(msg)

				if err != nil {
					log.Panic(err)
				}
				send(url+"/preprepare", jsonMsg)
			}
		case *PrepareMsg:
			for peerID, url := range p.peerList {
				//防止发消息给自己
				if peerID == p.peerID {
					continue
				}

				jsonMsg, err := json.Marshal(msg)

				if err != nil {
					log.Panic(err)
				}

				send(url+"/prepare", jsonMsg)
			}
		case *CommitMsg:
			for peerID, url := range p.peerList {
				//防止发消息给自己
				if peerID == p.peerID {
					continue
				}

				jsonMsg, err := json.Marshal(msg)

				if err != nil {
					log.Panic(err)
				}

				send(url+"/commit", jsonMsg)
			}
		}
	}

}

//广播请求给其他节点
func send(url string, msg []byte) {
	buff := bytes.NewBuffer(msg)
	http.Post("http://"+url, "application/json", buff)
}

//判断这个消息是否已经处理

//能处理哪些消息
func (p *Peer) handleMessage(msg interface{}) {
	switch et := msg.(type) {
	//这里的请求消息指的就是区块
	case *RequestMsgs:
		p.getReq(et)
	}
}

func NewServer(peerID int64) *Server {
	peer := NewPeer(peerID)
	server := &Server{peer.peerList[peerID], peer}
	server.setRoute()
	return server
}

func (server *Server) Start() {
	fmt.Printf("Server will be started at %s...\n", server.url)
	if err := http.ListenAndServe(server.url, nil); err != nil {
		log.Panic(err)
		return
	}
}

func (server *Server) setRoute() {
	http.HandleFunc("/req", server.getReq)
	http.HandleFunc("/preprepare", server.getPrePrepare)
	http.HandleFunc("/prepare", server.getPrepare)
	http.HandleFunc("/commit", server.getCommit)
	//http.HandleFunc("/reply", server.getReply)
}


func (server *Server) getReq(writer http.ResponseWriter, request *http.Request) {
	var msg RequestMsg
	//fmt.Print(1)
	err := json.NewDecoder(request.Body).Decode(&msg)
	if err != nil {
		fmt.Println(err)
		return
	}
	//to do 需要消息达到几百或者多少个,目前就一个
	msgs := RequestMsgs{
		reqs: []*RequestMsg{&msg},
	}

	server.peer.MsgEntrance <- &msgs
}

func (server *Server) getPrePrepare(writer http.ResponseWriter, request *http.Request) {
	var msg PrePrepareMsg
	//将接受到的消息转化为一个decoder，再将这个decode解码到msg
	err := json.NewDecoder(request.Body).Decode(&msg)
	if err != nil {
		fmt.Println(err)
		return
	}
	fmt.Println("预准备消息来了")
	server.peer.MsgEntrance <- &msg
}

func (server *Server) getPrepare(writer http.ResponseWriter, request *http.Request) {
	var msg PrepareMsg
	err := json.NewDecoder(request.Body).Decode(&msg)
	if err != nil {
		fmt.Println(err)
		return
	}

	server.peer.MsgEntrance <- &msg
}

func (server *Server) getCommit(writer http.ResponseWriter, request *http.Request) {
	fmt.Println("收到commit消息了")
	var msg CommitMsg
	err := json.NewDecoder(request.Body).Decode(&msg)
	if err != nil {
		fmt.Println(err)
		return
	}

	server.peer.MsgEntrance <- &msg
}

//创建新的节点
func NewPeer(peerID int64) *Peer {
	const viewID = 100
	peer := &Peer{
		f:      1,
		N:      4,
		peerID: peerID,
		k:      100,
		reqNo:  0,
		peerList: map[int64]string{
			0: "localhost:1111",
			1: "localhost:1112",
			2: "localhost:1113",
			3: "localhost:1114",
		},
		view:        viewID,
		reqStore:    make(map[string]*RequestMsgs),
		msgStore:    make(map[msgIndex]*msgSave),
		lastExecNum: 0,
		MsgEntrance: make(chan interface{}),
		MsgDelivery: make(chan interface{}),
		h:           0,
		H:           200,
	}

	//监听收到的消息
	go peer.dispatchMsg()

	//用来处理消息
	//go peer.handleMsg()

	go peer.Broadcast()

	return peer
}

//广播消息
func (node *Peer) dispatchMsg() {
	for {
		select {
		//接收到新的消息之后立即处理
		case msg := <-node.MsgEntrance:
			//路由处理
			err := node.routeMsg(msg)
			if err != nil {
				fmt.Println(err)
				// TODO: send err to ErrorChannel
			}
		}
	}
}

//func (p Peer) handleMsg() {
//	for {
//		msgs := <- p.MsgDelivery
//		switch msgs.(type) {
//			case []*RequestMsg:
//				errs := p.get
//		}
//	}
//}

func (p *Peer) routeMsg(msg interface{}) []error {
	switch ms := msg.(type) {
	case *RequestMsgs:
		//消息加入处理消息的通道
		fmt.Println("处理客户端请求消息")
		p.getReq(ms)
	case *PrePrepareMsg:
		p.getPrePrepareMsg(ms)
	case *PrepareMsg:
		p.getPrepareMsg(ms)
	case *CommitMsg:
		p.getCommitMsg(ms)
	}

	//to do错误处理
	return nil
}

//收到2f+1个相同的消息后确认共识
func (p *Peer) ConfirmConsensus(n int) bool {
	//return n >= (p.N + p.f + 1)/2
	return false
}

//新请求处理
func (p *Peer) getReq(reqMsgs *RequestMsgs) {
	//hash算法对接受到的多个请求进行摘要
	digest := util.Hash(reqMsgs)
	//存储接受到的所有区块
	p.reqStore[digest] = reqMsgs
	//请求处理

	//当前请求时主节点,发送pre-prepare消息，否则不进行任何操作
	if p.primaryID(p.view) == p.peerID {
		fmt.Println("我是主节点")
		p.sendPrePrepare(reqMsgs, digest)
	}
}

//节点接受到批量请求后判断自己分配序列号逻辑
func (p *Peer) sendPrePrepare(reqMsg *RequestMsgs, digest string) {

	//在判断了自己是不是主节点之后决定是否发送消息，首先也要进行判断自己之前是否接受到序列号为n,摘要不同的消息
	n := p.reqNo + 1
	for _, msgSave := range p.msgStore {
		if msgPrePrepare := msgSave.prePrepare; msgPrePrepare != nil {
			//判断当前节点之前是否接受到过该消息
			if p.view == msgPrePrepare.ViewID && p.reqNo != n && msgPrePrepare.Digest == digest && digest != "" {
				return
			}
		}
	}
	//请求编号是否在高低水位之间,这里是否需要控制一下检查点消息
	if n > p.H && n < p.h {
		return
	}
	//验证本节点是否收到过相同的编号，但是摘要不同的pre-prepare请求，如果是拒绝处理

	//给请求分配序号，这里需不要指定节点当前处理的消息序列号，应该需要指定
	//p.reqNo = n
	//发送pre-prepare消息
	prePrepare := &PrePrepareMsg{
		ViewID:     p.view,
		SequenceID: n,
		Digest:     digest,
		RequestMsg: reqMsg,
		PeerID:     p.peerID,
	}
	//主节点将预准备消息存入日志，根据v和n获取当前的消息
	msg := p.getMsgSave(p.view, n)
	msg.prePrepare = prePrepare
	msg.digest = digest
	fmt.Println("广播请求")
	//广播请求
	p.MsgDelivery <- prePrepare
	//
}

//处理pre-prepare消息

func (p *Peer) getPrePrepareMsg(prepare *PrePrepareMsg) {

	//是否处于视图更新阶段

	//是不是主节点发送的消息
	if p.primaryID(p.view) != prepare.PeerID {
		log.Panic("preparemsg is not send by primary")
		return
	}
	//请求的序列号是不是处于高低水线之间
	//请求编号是否在高低水位之间,这里是否需要控制一下检查点消息
	if prepare.SequenceID > p.H || prepare.SequenceID < p.h {
		log.Panic("消息不在高低水线之间")
		return
	}
	//view-change消息触发

	//如果之前的消息存储中没有收到过该消息，而且不是一个空消息
	msg := p.getMsgSave(prepare.ViewID, prepare.SequenceID)
	if msg.digest != "" && msg.digest != prepare.Digest {
		//触发view-change消息
		log.Panic("收到了相同的v,n但是摘要不同的消息")
		return
	}
	//将受到的新消息保存
	msg.prePrepare = prepare
	msg.digest = prepare.Digest
	//之前收到了该序列号，但是摘要不同的消丢弃该请求，没有该序列号的请求保存该消息

	//判断是否完成了pre-prepare阶段，这里也是需要判断一个消息是否达到了pre-prepare状态，同时prepare消息未发送
	if p.primaryID(p.view) != p.peerID && p.prePrepared(prepare.Digest, prepare.ViewID, prepare.SequenceID) && !msg.sentPrepare {
		//创建prepared消息
		prepareMsg := &PrepareMsg{
			ViewID:      prepare.ViewID,
			SequenceID:  prepare.SequenceID,
			BatchDigest: prepare.Digest,
			ReplicaId:   p.peerID,
		}
		//当确认可以发送prepared消息之后就可以将状态改为true
		msg.sentPrepare = true
		//更新已经完成的prepared请求

		//广播之前需要自己将收到的preprepare保存到自己的消息记录，也就是要添加到自己的msgSave和msgSave.prepare列表中
		//因为主节点不会处理pre-prepare消息，不会广播prepareMsg消息，也就是他不会在这里执行这个函数，那么他的msgStore列表中
		//就会少一个，主节点只是将消息存储为pre-prepare状态，并不会将消息更新为prepare状态，主节点和各自的备份节点都会少收到一个prepare消息
		//因此判断是不是达成prepare状态，在代码中体现是2f，并不是2f+1，但是2f就代表了2f+1个消息
		p.getPrepareMsg(prepareMsg)
		//这里其实有一个机制,因为主节点判断消息
		fmt.Println("节点开始广播准备消息")
		//广播消息
		p.MsgDelivery <- prepareMsg
	}

}

//节点处理prepare消息
func (p *Peer) getPrepareMsg(prep *PrepareMsg) {
	//准备消息不能是主节点发送
	if p.primaryID(prep.ViewID) == prep.ReplicaId {
		log.Panic("消息是主节点发送的忽略")
		return
	}

	//检查消息是否在高低水线之间
	if prep.SequenceID > p.H || prep.SequenceID < p.h {
		fmt.Println("消息序列号要处于高低水线之间")
		return
	}

	//检查prepared消息日志中是否保存过该消息，
	msg := p.getMsgSave(prep.ViewID, prep.SequenceID)
	//判断发送消息的节点是不是已经发送过该消息
	for _, prepareMsg := range msg.prepare {
		//如果该消息已经存储，而且发送消息的节点之前已经发送过消息
		if prepareMsg.ReplicaId == prep.ReplicaId {
			log.Panic("当前节点之前已经收到过同一节点的消息")
			return
		}
	}
	//如果不是上面那种情况，说明必须将收到的消息添加到存储的准备消息列表中，也就是那2f+1个消息存储中。
	msg.prepare = append(msg.prepare, prep)
	//消息存储到

	//把收到的消息保存在日志.所有这个保存消息的类型可能保存有prePrepare消息和prepare消息数组 []*prepare消息

	//判断是否可以发送commit消息
	p.maybeSendCommit(prep.BatchDigest, prep.ViewID, prep.SequenceID)
	//发送commit消息
	//1.验证是否发送过消息，请求完成prepared阶段，
	//验证是否完成prepared消息步骤
	//1.验证该消息是否完完成pre-prepare阶段
	//2.验证是否完成prepare阶段
	//cert中要找到对应的记录
	//统计当前节点收到的该请求的prepare消息个数是否大于2f+1

}

func (p *Peer) getCommitMsg(commit *CommitMsg) {

	//检查请求的编号是否在高低水线之间
	if commit.SequenceID > p.H || commit.SequenceID < p.h {
		return
	}
	fmt.Println("判断commit完成之前的日志")
	//检查prepared消息日志中是否保存过该消息，
	msg := p.getMsgSave(commit.ViewID, commit.SequenceID)
	//判断发送消息的节点是不是已经发送过该消息
	for _, commitMsg := range msg.commit {
		//如果该消息已经存储，而且发送消息的节点之前已经发送过消息
		if commitMsg.ReplicaId == commit.ReplicaId {
			log.Panic("当前节点之前已经收到过同一节点的消息")
			return
		}
	}
	//不是上面那种情况,把消息添加到从其他节点收到的commit消息列表
	msg.commit = append(msg.commit,commit)
	fmt.Println("判断commit完成之前")
	//判断是否完成commit阶段
	if p.commited(commit.Digest, commit.ViewID, commit.SequenceID) {
		fmt.Println("commit阶段已完成，开始执行请求")
		//执行请求
	}
	//记录收到的commt消息

	//请求已经完成commit阶段，一

}

//视图更改协议， 除了从节点主动触发视图切换，PBFT还会定时的更新主节点，

//
